package com.vichat.common.rudp;
import com.vichat.common.rudp.thread.RudpBusiThread;
import com.vichat.common.rudp.thread.RudpNioReceiveManager;
import com.vichat.common.rudp.thread.RudpReceiveManager;
import com.vichat.common.rudp.thread.RudpResponseThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;


public class RudpDataUtil {
    private static transient Logger logger = LoggerFactory.getLogger(RudpDataUtil.class);

    public static ConcurrentHashMap<String, HashMap> synreqData = new ConcurrentHashMap<String, HashMap>();
    public static HashMap<String, RudpTransData> transDataCont = new HashMap<String, RudpTransData>();
    //public static HashMap<String,TransLocker> transLockerCont = new HashMap<byte[],TransLocker>();
    public static HashMap<String, RudpBlockData> sendDataCont = new HashMap<String, RudpBlockData>();
    public static Object sendLocker = new Object();
    public static int maxBlockSize = 900;
    public static int transseq = 0;
    public static Object socketUserLocker = new Object();
    private static int serverMode = 1;//1 nio 0 block
    private static SocketAddress bindAddress = null;


    private static DatagramSocket serverSocket = null;
    private static DatagramChannel serverSocketNio = null;

    private static Object socketLocker = new Object();

    private static Object transDataLocker = new Object();

    private static Random gloRandom = new Random();


    private static ExecutorService busiExec = null;
    private static RudpBusiSrv busiService = null;
    private static RudpResponseSrv respService = null;
    private static int busiType = 0;
    private static byte predata = 58;

    public static void initServerMode(int inserverMode) {
        serverMode = inserverMode;
    }

    public static void initSocket(SocketAddress inaddress) {
        if (serverMode == 1) {
            initSocketNio(inaddress);
        } else {
            initSocketObj(inaddress);
        }
    }

    public static void initSocketObj(SocketAddress inaddress) {
        bindAddress = inaddress;
        try {
            serverSocket = new DatagramSocket(bindAddress);
            serverSocket.setReceiveBufferSize(1024 * 1024 * 2);
            serverSocket.setSendBufferSize(1024 * 1024 * 2);
            logger.debug("SOCKET init succ:" + bindAddress.toString());
        } catch (SocketException e) {
            logger.error("SOCKET init err:" + bindAddress.toString(), e);
        }
    }

    public static void initSocketNio(SocketAddress inaddress) {
        bindAddress = inaddress;
        try {
            serverSocketNio = DatagramChannel.open();
            serverSocketNio.configureBlocking(false);
            serverSocket = serverSocketNio.socket();
            serverSocket.bind(bindAddress);
            serverSocket.setReceiveBufferSize(1024 * 1024 * 2);
            serverSocket.setSendBufferSize(1024 * 1024 * 2);
            logger.debug("NIOSOCKET init succ:" + bindAddress.toString());
        } catch (Exception e) {
            logger.error("NIOSOCKET init err:" + bindAddress.toString(), e);
        }
    }

    public static void initBusiObj(ExecutorService inexec, RudpBusiSrv inservice, RudpResponseSrv inresp, int intype) {
        busiExec = inexec;
        busiService = inservice;
        busiType = intype;
        respService = inresp;
    }

    public static void startReceiver() throws Exception {
        if (serverMode == 1) {
            RudpNioReceiveManager receiveManager = new RudpNioReceiveManager();
            receiveManager.start();
            logger.debug("RUDP nioreceiver started");
        } else {
            RudpReceiveManager receiveManager = new RudpReceiveManager();
            receiveManager.start();
            logger.debug("RUDP receiver started");
        }
    }

    public static DatagramChannel getServerSocketNio() {
        if (serverSocketNio != null && serverSocketNio.isOpen()) {
            return serverSocketNio;
        }
        synchronized (socketLocker) {
            if (serverSocketNio == null || !serverSocketNio.isOpen()) {
                try {
                    serverSocketNio = DatagramChannel.open();
                    serverSocketNio.configureBlocking(false);
                    serverSocket = serverSocketNio.socket();
                    serverSocket.bind(bindAddress);
                    serverSocket.setReceiveBufferSize(1024 * 1024 * 2);
                    serverSocket.setSendBufferSize(1024 * 1024 * 2);
                    logger.debug("NIOSOCKET bind succ:" + bindAddress.toString());
                } catch (Exception e) {
                    logger.error("NIOSOCKET bind error:" + bindAddress.toString(), e);
                }
            }

        }
        return serverSocketNio;
    }

    public static DatagramSocket getServerSocket() {
        if (serverSocket != null && !serverSocket.isClosed()) {
            return serverSocket;
        }
        synchronized (socketLocker) {
            if (serverSocket == null || serverSocket.isClosed()) {
                try {
                    serverSocket = new DatagramSocket(bindAddress);
                    serverSocket.setReceiveBufferSize(1024 * 1024 * 2);
                    serverSocket.setSendBufferSize(1024 * 1024 * 2);
                    logger.debug("SOCKET bind succ:" + bindAddress.toString());
                } catch (SocketException e) {
                    logger.error("SOCKET bind error:" + bindAddress.toString(), e);
                }
            }

        }
        return serverSocket;
    }

    public static void getTransLocker(byte[] transId) {
        //if(!transLockerCont.containsKey(transId)){
        //ransLocker transtime = new TransLocker(transId,System.currentTimeMillis());
        //transLockerCont.put(transId, new Long(System.currentTimeMillis()));
        //return transtime;
        //}
    }


    public static synchronized RudpTransData assembleData(RudpBlockData blockData) throws Exception {
        RudpTransData transData = null;

        //synchronized(transLock){
        if (transDataCont.containsKey(blockData.getStringTrans())) {
            if (blockData.blockNum > 1) {
                logger.debug("多宝处理,已有事务对象" + blockData.blockId);
            }
            transData = transDataCont.get(blockData.getStringTrans());

            if (transData.indicated) {
                throw new Exception("包已收全,过期无需处理");
            }
            if (transData.blockNum == 1) {
                throw new Exception("包已收到,重复无需处理");
            } else {
                logger.debug("多包组装" + blockData.getStringTrans() + blockData.blockId);
            }
            //判断是否已经有包
            for (Iterator<RudpBlockData> iter = transData.blockList.iterator(); iter.hasNext(); ) {
                RudpBlockData tmpdata = iter.next();
                if (tmpdata.blockId == blockData.blockId) {
                    throw new Exception("包已收到,重复无需处理");
                }
            }
            transData.blockList.add(blockData);
        } else {
            if (blockData.blockNum > 1) {
                logger.debug("多宝处理,没有事务对象" + blockData.getStringTrans() + "-" + blockData.blockId);
            }
            transData = new RudpTransData(blockData.transId, blockData.blockNum, blockData.lifeTime);
            transData.blockList.add(blockData);
            synchronized (transDataLocker) {
                transDataCont.put(blockData.getStringTrans(), transData);
            }
        }
        transData.receiveNum++;
        transData.dataLength += blockData.blockLen;
        //}
        return transData;

    }

    public static HashMap getTransCloneData() {
        synchronized (transDataLocker) {
            return (HashMap) RudpDataUtil.transDataCont.clone();
        }
    }

    public static void removeTransData(String transId) {
        synchronized (transDataLocker) {
            RudpDataUtil.transDataCont.remove(transId);
        }
    }

    public static ParseData assembelBusiData(RudpTransData transData) {
        if (transData.blockNum > 1) {
            logger.debug("多包组装开始");
        }
        byte[] destarray = new byte[transData.dataLength];
        for (Iterator<RudpBlockData> iter = transData.blockList.iterator(); iter.hasNext(); ) {
            RudpBlockData blockObj = iter.next();
            int destPos = blockObj.blockId * maxBlockSize;
            System.arraycopy(blockObj.blockData, 0, destarray, destPos, blockObj.blockLen);
        }

        //去掉首位 :
        if (destarray[0] != predata) {
            throw new RuntimeException("报文错误");
        }
        byte[] sizearray = new byte[3];
        System.arraycopy(destarray, 1, sizearray, 0, 3);
        int datasize = ProtocolUtil.byte2int(sizearray);
        logger.debug("!!!!!报文预计长度" + datasize + ",实际长度" + destarray.length);
        if (datasize != destarray.length - 4) {
            logger.debug("!!!!!!!!报文预计长度和实际长度不符报文预计长度" + datasize + ",实际长度" + destarray.length);
        }
        byte[] destarray2 = new byte[destarray.length - 4];
        System.arraycopy(destarray, 4, destarray2, 0, destarray.length - 4);

        ParseData busiData = new ParseData(destarray2);
        if (transData.blockNum > 1) {
            logger.debug("多包组装完成" + busiData.toString());
        }
        return busiData;
    }


    public static RudpBlockData unmarshalBlockData(byte[] rudpbytes) throws Exception {
        RudpBlockData blockData = new RudpBlockData();
        byte[] tmpByteArray = new byte[10];
        blockData.msgType = (int) rudpbytes[0];
        if (blockData.msgType == 1) {
            System.arraycopy(rudpbytes, 1, tmpByteArray, 0, 2);
            blockData.blockId = ByteUtil.byte2int(tmpByteArray, 2);
            System.arraycopy(rudpbytes, 3, tmpByteArray, 0, 10);
            blockData.transId = tmpByteArray;
        } else {
            blockData.lifeTime = (int) rudpbytes[1];
            System.arraycopy(rudpbytes, 2, tmpByteArray, 0, 2);
            blockData.blockNum = ByteUtil.byte2int(tmpByteArray, 2);
            System.arraycopy(rudpbytes, 4, tmpByteArray, 0, 2);
            blockData.blockId = ByteUtil.byte2int(tmpByteArray, 2);
            System.arraycopy(rudpbytes, 6, tmpByteArray, 0, 2);
            blockData.blockLen = ByteUtil.byte2int(tmpByteArray, 2);
            System.arraycopy(rudpbytes, 8, tmpByteArray, 0, 10);
            blockData.transId = tmpByteArray;
            //处理业务报文
            int maxlen = rudpbytes.length - 18;
            /*if (blockData.blockLen > maxlen) {
                throw new Exception("业务数据错误,预期长度" + blockData.blockLen + ",最大长度" + maxlen);
            }*/
            tmpByteArray = new byte[maxlen];
            System.arraycopy(rudpbytes, 18, tmpByteArray, 0, maxlen);
            blockData.blockData = tmpByteArray;
        }
        return blockData;
    }

    public static byte[] marshalBlockData(RudpBlockData blockData) {
        int datalen = blockData.blockLen + 18;
        byte[] rudpbytes = new byte[datalen];
        rudpbytes[0] = (byte) blockData.msgType;
        rudpbytes[1] = (byte) blockData.lifeTime;
        System.arraycopy(ByteUtil.int2byte(blockData.blockNum, 2), 0, rudpbytes, 2, 2);
        System.arraycopy(ByteUtil.int2byte(blockData.blockId, 2), 0, rudpbytes, 4, 2);
        System.arraycopy(ByteUtil.int2byte(blockData.blockLen, 2), 0, rudpbytes, 6, 2);
        System.arraycopy(blockData.transId, 0, rudpbytes, 8, 10);
        System.arraycopy(blockData.blockData, 0, rudpbytes, 18, blockData.blockLen);
        return rudpbytes;
    }


    public static byte[] marshalAckData(RudpBlockData ackData) {
        byte[] rudpbytes = new byte[13];
        rudpbytes[0] = (byte) ackData.msgType;
        System.arraycopy(ByteUtil.int2byte(ackData.blockId, 2), 0, rudpbytes, 1, 2);
        System.arraycopy(ackData.transId, 0, rudpbytes, 3, 10);
        return rudpbytes;
    }

    public static void receiveAckPacket(RudpBlockData ackData) throws Exception {

        if (RudpDataUtil.sendDataCont.containsKey(ackData.getBlockKey())) {

            synchronized (sendLocker) {
                RudpDataUtil.sendDataCont.remove(ackData.getBlockKey());
            }
            if (RudpMonitorUtil.monitorFlag) {
                RudpMonitorUtil.receiveAck();
            }
            //RudpBlockData sendData = RudpDataUtil.sendDataCont.get(ackData.getBlockKey());
            //sendData.lifeTime = -1;
            logger.error("有效的ACK的key" + ackData.getBlockKey());
        } else {
            if (RudpMonitorUtil.monitorFlag) {
                RudpMonitorUtil.errorAck();
            }

            logger.error("接受到的ACK无效" + ackData.getStringTrans() + "-" + ackData.getBlockKey());
        }
    }


    public static void sendAckPacket(RudpBlockData ackData) throws Exception {
        try {
            if (serverMode == 0) {
                byte[] buffer = new byte[1024]; // 缓冲区
                DatagramPacket packet = new DatagramPacket(buffer, buffer.length, ackData.desAddress);
                packet.setData(ackData.toPocketArray());
                synchronized (socketUserLocker) {
                    getServerSocket().send(packet);
                }

                //logger.error("发送ACK成功"+ackData.transId+"-"+ackData.blockId+";"+ackData.desAddress.toString()+";"+new String(ackData.toPocketArray()));
            } else {
                ByteBuffer buffer = ByteBuffer.wrap(ackData.toPocketArray());
                //serverSocketNio
                serverSocketNio.send(buffer, ackData.desAddress);
                //serverSocketNio.write(src)(buffer, blockData.desAddress);
            }
        } catch (Exception e) {
            logger.error("发送单个报文失败" + ackData.transId + "-" + ackData.blockId + ";" + ackData.desAddress.toString() + ";" + ackData.getBlockKey(), e);
        }
    }

    public static void sendBlockData(RudpBlockData blockData) {
        try {
            if (serverMode == 0) {
                byte[] buffer = new byte[1024];
                DatagramPacket packet = new DatagramPacket(buffer, buffer.length, blockData.desAddress);
                packet.setData(blockData.toPocketArray());
                synchronized (socketUserLocker) {
                    getServerSocket().send(packet);
                }
            } else {
                ByteBuffer buffer = ByteBuffer.wrap(blockData.toPocketArray());
                //serverSocketNio
                serverSocketNio.send(buffer, blockData.desAddress);
                //serverSocketNio.write(src)(buffer, blockData.desAddress);
            }
            logger.debug("发送数据成功"+blockData.transId+"-"+blockData.blockId+";"+blockData.desAddress.toString()+",数据长度"+blockData.toPocketArray().length);
        } catch (Exception e) {
            logger.error("发送数据失败" + blockData.transId + "-" + blockData.blockId + ";" + blockData.desAddress.toString() + ";" + new String(blockData.blockData), e);
        }
    }

    public static HashMap<String, RudpBlockData> getResendDate() {
        synchronized (sendLocker) {
            return (HashMap<String, RudpBlockData>) sendDataCont.clone();
        }
    }

    public static void addResendData(RudpBlockData packet) {
        String resendkey = packet.getBlockKey();
        logger.debug("重发数据主键" + resendkey);
        synchronized (sendLocker) {
            if (!RudpDataUtil.sendDataCont.containsKey(resendkey)) {
                RudpDataUtil.sendDataCont.put(resendkey, packet);
            }
        }
    }

    public static void delResendData(String blockKey) {
        synchronized (sendLocker) {
            sendDataCont.remove(blockKey);
        }
    }

    public static void addBusiQueue(RudpTransData data) {
        long curtime = 0;
        if (RudpMonitorUtil.monitorFlag) {
            curtime = System.currentTimeMillis();
        }
        if (busiType == 0) {
            busiExec.execute(new RudpBusiThread(busiService, data));
        } else {
            busiExec.execute(new RudpResponseThread(respService, data));
        }
        if (RudpMonitorUtil.monitorFlag) {
            //统计开始进入业务
            RudpMonitorUtil.busiTrans(data.getStringTrans(), curtime);
            RudpMonitorUtil.startBusi();
        }
    }


    public static synchronized byte[] getNewTransId(long curtime) {
        StringBuilder transbd = new StringBuilder(10);
        if (curtime >= 100 && curtime < 1000) {
            transbd.append(curtime);
        } else {
            transbd.append(String.valueOf(curtime + 1000).substring(0, 3));
        }

        if (transseq < 10) {
            transbd.append("0000");
        } else if (transseq < 100) {
            transbd.append("000");
        } else if (transseq < 1000) {
            transbd.append("00");
        } else if (transseq < 10000) {
            transbd.append("0");
        }
        transbd.append(transseq);
        if (transseq >= 99999) {
            transseq = 0;
        } else {
            transseq++;
        }
        transbd.append(gloRandom.nextInt(9));
        transbd.append(gloRandom.nextInt(9));
        //transbd.append(gloRandom.nextInt(9));
        return transbd.toString().getBytes();
    }

    /**
     * 同步发送数据,等待返回结果
     */
    public static Map<Integer, String> sendDataSync(HashMap<Integer, String> dataobj, SocketAddress destAddress, String respTrans) throws Exception {

        byte[] seqByte = getNewTransId(System.currentTimeMillis() / 1000);
        String seqId = ProtocolUtil.transIdStr(seqByte);
        dataobj.put(RudpParamDefine.cmd_param_crypt_data, seqId);
        synreqData.put(seqId,new HashMap<Integer,String>());
        logger.error("等待返回数据seqId=" + seqId);
        sendDataAsyn(dataobj, destAddress, respTrans);

        for (int i = 0; i < 10; i++) {
            if (synreqData.containsKey(seqId) && synreqData.get(seqId) != null) {
                Map<Integer, String> reqObj = synreqData.get(seqId);
                if(reqObj.size()>0){
                    logger.error("发送数据返回结果：" + reqObj.toString());
                    return reqObj;
                }
            }
            Thread.sleep(1000);
        }

        return null;
    }

    public static void putSendResponse(String seqid, HashMap<Integer, String> dataresp){
        if(synreqData.containsKey(seqid) && synreqData.get(seqid) == null){
            synreqData.put(seqid, dataresp);
        }
    }


    //异步发送数据,不等待返回结果
    public static void sendDataAsyn(HashMap<Integer, String> dataobj, SocketAddress destAddress, String respTrans) throws Exception {

        byte[] prearray = ProtocolUtil.encodeMap(dataobj);
        byte[] dataarray = new byte[prearray.length + 4];
        dataarray[0] = predata;
        byte[] sizearray = new byte[3];
        sizearray = ProtocolUtil.int2sizebyte(prearray.length);
        System.arraycopy(sizearray, 0, dataarray, 1, 3);
        System.arraycopy(prearray, 0, dataarray, 4, prearray.length);

        int blocknum = dataarray.length / maxBlockSize;
        int lastBlocknum = 0;
        if (dataarray.length % maxBlockSize > 0) {
            blocknum++;
            lastBlocknum = dataarray.length % maxBlockSize;
        }

        //byte[] buffer = new byte[1024]; 
        long curMisec = System.currentTimeMillis();
        long cuttime = curMisec % 1000;
        byte[] transId = getNewTransId(cuttime);
        for (int bi = 0; bi < blocknum; bi++) {
            int blocklen = maxBlockSize;
            if (bi == blocknum - 1 && lastBlocknum > 0) {
                blocklen = lastBlocknum;
            }
            byte[] tmpdata = new byte[blocklen];
            int startpos = bi * maxBlockSize;
            System.arraycopy(dataarray, startpos, tmpdata, 0, blocklen);
            RudpBlockData blockData = new RudpBlockData(bi, transId, 10, blocknum, blocklen, tmpdata);
            blockData.desAddress = destAddress;
            /*byte[] blockPacket = marshalBlockData(blockData);
            DatagramPacket packet = new DatagramPacket(buffer, buffer.length, destAddress, destPort);
            packet.setData(blockPacket);
            blockData.desPacket = packet;*/
            addResendData(blockData);
            blockData.firstTime = curMisec;
            blockData.lastTime = curMisec;
            sendBlockData(blockData);
            if (RudpMonitorUtil.monitorFlag) {
                RudpMonitorUtil.sendBlock();
            }
        }

    }

    public static void main(String[] args) {

        char test = ':';
        System.out.println((byte) test);
        //long curMisec = System.currentTimeMillis();
        //long cuttime = curMisec%1000;
        //System.out.print(RudpDataUtil.getNewTransId(cuttime));
        //System.out.print(RudpDataUtil.getNewTransId(cuttime));
    }

}
